//
// Created by 郝羽 on 2018/5/15.
//

#include "remote_connector.hpp"

#include <gio/gio.h>
#include <list>
#include <map>
#include <arpa/inet.h>

#include "placeholder.hpp"
#include "rtcp_utility.h"
#include "rtp.hpp"
#include "xcutil.h"
#include "xm_app_config.h"


struct remote_connector;

struct remote_linker {
    int64_t last_update_ms;
    remote_connector* connector;
    rtc_endpoint_t rtc;
    bool audio;
    bool video;
    rtc_endpoint_callback_link_t callback_link;
};

struct remote_connector {
    int64_t last_update_ms;
    GSocket* sock;
    std::map<std::string, remote_linker*> linkers;
    std::string rtc_id;
    GSocketAddress* remote_addr;
    guint sock_in_source_id;
    std::vector<uint32_t> peer_audio_ssrcs; // pub rtc local ssrc
    std::vector<uint32_t> peer_video_ssrcs;
};

struct remote_connector_manager {
    std::map<std::string, remote_connector*> connectors;
    GSource* timeout_source;
    const char* signal_ip;
};

static gboolean _rcm_handle_timeout(gpointer user_data);

static remote_connector* _rcm_new_connector(const char *rtc_id);

static void _rcm_connector_delete(remote_connector *connector);

static int _rcm_connector_count_linker(remote_connector *connector);

static gboolean _rcm_connector_handle_recv(GIOChannel *source, GIOCondition cond, gpointer data);

static gboolean _rcm_connector_handle_recv_rtcp(remote_connector *connector, const unsigned char *buf, int size);

static gboolean _rcm_connector_handle_recv_rtp(remote_connector *connector, const unsigned char *buf, int size);

static void _rcm_connector_send_rtcp(remote_connector* connector, const unsigned char* buf, int size);

static remote_connector* rcm_get_connector(remote_connector_manager* rcm, const std::string& rtc_id);

static void _rcm_linker_delete(remote_linker *linker);

static void _rcm_linker_callback_rtc_destroy(rtc_endpoint_t rtc, void * context);

static bool _rcm_linker_callback_on_data(rtc_endpoint_t rtc, void * context, int stream_type, rtc_codec_id_t codec, rtc_codec_id_t fec_codec, int data_type, int is_rtcp, int is_rtx, unsigned char * buf, int len);

constexpr int rcm_connector_timeout = 60000;
constexpr int rcm_linker_timeout = 60000;
constexpr int rcm_check_timeout = 1000;

rtc_endpoint_callbacks _rmc_linker_callbacks {
        .context = nullptr,
        .on_destroy = _rcm_linker_callback_rtc_destroy,
        .on_stream_data = _rcm_linker_callback_on_data,
        .on_event = nullptr,
};

remote_connector_manager* rcm_new(GMainContext *ctx) {

    auto rcm = new remote_connector_manager();

    rcm->timeout_source = g_timeout_source_new(rcm_check_timeout);

    g_source_set_callback(rcm->timeout_source, _rcm_handle_timeout, rcm, nullptr);

    g_source_attach(rcm->timeout_source, ctx);

    return rcm;
}

void rcm_del(remote_connector_manager *rcm) {

    auto it = rcm->connectors.begin();

    while (it != rcm->connectors.end()) {
        auto it_cur = it++;

        auto connector = it_cur->second;

        _rcm_connector_delete(connector);

        rcm->connectors.erase(it_cur);
    }

    g_source_remove(g_source_get_id(rcm->timeout_source));

    g_source_unref(rcm->timeout_source);

    delete(rcm);
}

linker_status rcm_add_linker(remote_connector_manager* rcm,
        const char* pub_rtc_id, rtc_endpoint_t sub_rtc,
        std::vector<uint32_t> &peer_audio_ssrcs,
        std::vector<uint32_t> &peer_video_ssrcs) {
    linker_status st;

    auto connector = rcm_get_connector(rcm, pub_rtc_id);

    if (connector == nullptr) {
        dbge("rcm_add_linker: remote connector '%s' add linker '%s', error: not found connector", pub_rtc_id, rtc_endpoint_get_id(sub_rtc));

        st.status = 500;
        st.audio = false;
        st.video = false;
        st.update = false;
        return st;
    }

    // 通过日志发现存在重复调用subR的情况，重复调用subR将会导致 crash
    // crash原因： 每次调用 rcm_add_linker 生成一个linker，并添加到rtc->listeners中；但是 connector map存储了最后一个；
    // 在 destory 或 unsubR后，connector会被释放，然后rtc destory，回调用 link->callbacks, 从而导致crash
    if(connector->linkers.find(rtc_endpoint_get_id(sub_rtc)) != connector->linkers.end()){
        dbge("rcm_add_linker: remote connector '%s' repeat add linker '%s', error", pub_rtc_id, rtc_endpoint_get_id(sub_rtc));

        st.status = 500;
        st.audio = false;
        st.video = false;
        st.update = false;
        return st;
    }

    bool connector_ssrc_updated = false;
    // audio_ssrcs must be first1
    // see https://en.cppreference.com/w/cpp/algorithm/equal
    if (peer_audio_ssrcs.size() != connector->peer_audio_ssrcs.size()
            || !std::equal(peer_audio_ssrcs.begin(), peer_audio_ssrcs.end(),
            connector->peer_audio_ssrcs.begin())) {
        connector->peer_audio_ssrcs.clear();
        connector->peer_audio_ssrcs = peer_audio_ssrcs; //pub rtc local ssrc
        connector_ssrc_updated = true;
    }

    if (peer_video_ssrcs.size() != connector->peer_video_ssrcs.size()
            || !std::equal(peer_video_ssrcs.begin(), peer_video_ssrcs.end(),
            connector->peer_video_ssrcs.begin())) {
        connector->peer_video_ssrcs.clear();
        connector->peer_video_ssrcs = peer_video_ssrcs; //pub rtc local ssrc
        connector_ssrc_updated = true;
    }

    auto linker = new remote_linker();
    linker->connector = connector;
    linker->rtc = sub_rtc;
    linker->audio = !peer_audio_ssrcs.empty();
    linker->video = !peer_video_ssrcs.empty();
    linker->callback_link = rtc_endpoint_register_callback(sub_rtc, linker, &_rmc_linker_callbacks);

    int audio_count = 0;
    int video_count = 0;

    for (auto it : connector->linkers) {
        auto old_link = it.second;

        if (old_link->audio) {
            ++audio_count;
        }
        if (old_link->video) {
            ++video_count;
        }
    }

    connector->linkers.emplace(rtc_endpoint_get_id(sub_rtc), linker);

    st.status = 0;
    st.audio = audio_count > 0 || !peer_audio_ssrcs.empty();
    st.video = video_count > 0 || !peer_video_ssrcs.empty();
    st.update = false;

    if (audio_count == 0 && !peer_audio_ssrcs.empty()) {
        dbgi("rcm_add_linker: remote connector '%s' add linker '%s', first audio link", pub_rtc_id, rtc_endpoint_get_id(sub_rtc));
        st.update = true;
    }

    if (video_count == 0 && !peer_video_ssrcs.empty()) {
        dbgi("rcm_add_linker: remote connector '%s' add linker '%s', first video link", pub_rtc_id, rtc_endpoint_get_id(sub_rtc));
        st.update = true;
    }

    if (connector_ssrc_updated) {
        dbgw("rcm_add_linker: remote connector '%s' audio/video ssrc changed", pub_rtc_id);
        st.update = true;
    }

    return st;
}

linker_status rcm_del_linker(remote_connector_manager* rcm, const char* pub_rtc_id, rtc_endpoint_t sub_rtc) {
    linker_status st;

    auto cit = rcm->connectors.find(pub_rtc_id);

    if (cit == rcm->connectors.end()) {
        st.status = 404;
        st.audio = false;
        st.video = false;
        st.update = false;
        return st;
    }

    auto connector = cit->second;

    if (connector == nullptr) {
        st.status = 500;
        st.audio = false;
        st.video = false;
        st.update = false;
        return st;
    }

    int audio_count = 0;
    int video_count = 0;

    for (auto it : connector->linkers) {
        auto old_link = it.second;

        if (old_link->audio) {
            ++audio_count;
        }
        if (old_link->video) {
            ++video_count;
        }
    }

    st.status = 0;
    st.audio = audio_count>1;
    st.video = video_count>1;
    st.update = !st.audio && !st.video; // update will tell proxy to send upubR

    auto rtc_id = rtc_endpoint_get_id(sub_rtc);
    auto it = connector->linkers.find(rtc_id);
    if (it == connector->linkers.end()) {
        return st;
    }

    auto linker = it->second;

    connector->linkers.erase(rtc_id);

    _rcm_linker_delete(linker);

    if (connector->linkers.size() == 0) {
        dbgi("rcm_del_linker: remote connector '%s' has no linker, delete it", pub_rtc_id);
        rcm->connectors.erase(connector->rtc_id);
        _rcm_connector_delete(connector);

    }

    return st;
}

const char* rcm_get_address(remote_connector_manager* rcm, const char* rtc_id) {
    auto connector = rcm_get_connector(rcm, rtc_id);

    if (connector == nullptr) {
        return nullptr;
    }
    GError* err = nullptr;
    auto addr = g_socket_get_local_address(connector->sock, &err);
    if (err != nullptr) {
        g_error_free(err);
        return nullptr;
    }

    struct sockaddr_in native_addr;
    g_socket_address_to_native(addr, &native_addr, sizeof(native_addr), &err);
    g_object_unref(addr);

    if (err != nullptr) {
        g_error_free(err);
        return nullptr;
    }

    static char addr_str[32] = {0,};

    sprintf(addr_str, "%s:%d", inet_ntoa(native_addr.sin_addr), ntohs(native_addr.sin_port));

    return addr_str;
}

static gboolean _rcm_handle_timeout(gpointer user_data) {

    auto rcm = static_cast<remote_connector_manager*>(user_data);

    auto now = get_timestamp_ms();

    auto it = rcm->connectors.begin();

    while (it != rcm->connectors.end()) {

        auto it_cur = it++;

        auto connector_id = it_cur->first;

        auto connector = it_cur->second;

        if (now - connector->last_update_ms > rcm_connector_timeout) {

            time_t t = connector->last_update_ms / 1000;

            auto ts = localtime(&t);

            char tstr[64] = {0,};

            strftime(tstr, sizeof(tstr), "%a %Y-%m-%d %H:%M:%S %Z", ts);

            dbgi("_rcm_handle_timeout: remote connector '%s' timeout, last seen at '%s',  delete it", connector_id.c_str(), tstr);

            _rcm_connector_delete(connector);

            rcm->connectors.erase(it_cur);

            continue;
        }

        if (_rcm_connector_count_linker(connector) == 0) {

            dbgi("_rcm_handle_timeout: remote connector '%s' is empty, delete it", connector_id.c_str());

            _rcm_connector_delete(connector);

            rcm->connectors.erase(it_cur);
        }
    }

    return TRUE;
}

static remote_connector* _rcm_new_connector(const char *rtc_id) {

    GError* err = nullptr;
    GSocket* sock = g_socket_new(G_SOCKET_FAMILY_IPV4, G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err);

    if (err != nullptr) {
        dbge("_rcm_new_connector('%s'): could not create connector socket: %s", rtc_id, err->message);
        g_error_free(err);
        return nullptr;
    }

    g_socket_bind(sock, g_inet_socket_address_new(g_inet_address_new_from_string(app_config_get()->ipm_ip), 0), FALSE, &err);

    if (err != nullptr) {
        dbge("_rcm_new_connector('%s'): could not bind connector socket: %s", rtc_id, err->message);
        g_error_free(err);
        g_object_unref(sock);
        return nullptr;
    }

    auto bind_addr = g_socket_get_local_address(sock, &err);

    if (err != nullptr) {
        dbge("_rcm_new_connector('%s'): could not get bind address: %s", rtc_id, err->message);
        g_free(err);
        g_object_unref(sock);
        return nullptr;
    }

    sockaddr_in sa;

    g_socket_address_to_native(bind_addr, &sa, sizeof(sa), &err);
    g_object_unref(bind_addr);

    if (err != nullptr) {
        dbge("_rcm_new_connector('%s'): could not convert bind address to sockaddr_in", rtc_id);
        g_free(err);
        g_object_unref(sock);
        return nullptr;
    }

    dbgi("_rcm_new_connector('%s'): bind to '%s:%d'", rtc_id, inet_ntoa(sa.sin_addr), ntohs(sa.sin_port));

    g_socket_set_blocking(sock, FALSE);

    auto connector = new remote_connector();

    connector->last_update_ms = get_timestamp_ms();

    connector->sock = sock;

    connector->rtc_id = rtc_id; // pubRtcId

    connector->remote_addr = nullptr;

    auto ch = g_io_channel_unix_new(g_socket_get_fd(connector->sock));

    connector->sock_in_source_id = g_io_add_watch(ch, G_IO_IN, _rcm_connector_handle_recv, connector);

    g_io_channel_unref(ch);

    return connector;
}

static void _rcm_connector_delete(remote_connector *connector) {
    std::string rtc_id = connector->rtc_id;
    dbgi("_rcm_connector_delete('%s')", rtc_id.c_str());

    if (connector->sock_in_source_id > 0) {
        g_source_remove(connector->sock_in_source_id);
    }

    GError* err = nullptr;

    g_socket_close(connector->sock, &err);

    if (err != nullptr) {
        g_error_free(err);
    }

    g_object_unref(connector->sock);

    connector->sock = nullptr;

    auto it = connector->linkers.begin();

    while (it != connector->linkers.end()) {

        auto it_cur = it++;

        auto linker = it_cur->second;

        _rcm_linker_delete(linker);

        connector->linkers.erase(it_cur);
    }

    if (connector->remote_addr != nullptr) {
        g_object_unref(connector->remote_addr);
        connector->remote_addr = nullptr;
    }

    delete(connector);
    dbgi("_rcm_connector_delete('%s') end", rtc_id.c_str());
}

static int _rcm_connector_count_linker(remote_connector *connector) {
    return connector->linkers.size();
}

/**
 * 终端A--rtcp|rtp-> Xswitch1(pub_rtc) --socket:udp rtcp|rtp-> Xswitch2:_rcm_connector_handle_recv(sub_rtc) --rtcp|rtp->终端B
 * Xswitch1 将 rtcp pub rtc remote ssrc --> placeholder;  (process_recv_rtp_rtcp)
 *             rtp pub rtc remote ssrc --> pub rtc local ssrc (send_rtp_data)
 *
 *
 * @param source
 * @param conf
 * @param data
 * @return
 */
static gboolean _rcm_connector_handle_recv(GIOChannel *source, GIOCondition conf, gpointer data) {
    auto connector = static_cast<remote_connector*>(data);

    unsigned char buf[1600] = {0,};

    GError* err = nullptr;

    int size = 0;

    while (true) {
        if (connector->remote_addr == nullptr) {

            GSocketAddress* remote_addr = nullptr;

            size = g_socket_receive_from(connector->sock, &remote_addr, (gchar*)buf, sizeof(buf), nullptr, &err);

            if (err != nullptr) {
                if (err->code != G_IO_ERROR_WOULD_BLOCK) {
                    dbgw("_rcm_connector_handle_recv(%s): %s", connector->rtc_id.c_str(), err->message);
                }
                g_error_free(err);
                if (remote_addr) {
                    g_object_unref(remote_addr);
                }
                return G_SOURCE_CONTINUE;
            }

            sockaddr_in sa;
            g_socket_address_to_native(remote_addr, &sa, sizeof(sa), &err);

            if (err != nullptr) {
                dbgw("_rcm_connector_handle_recv(%s): %s", connector->rtc_id.c_str(), err->message);
                g_error_free(err);
                g_object_unref(remote_addr);
                return G_SOURCE_CONTINUE;
            } else {
                dbgi("_rcm_connector_handle_recv(%s): set remote address to '%s:%d'", connector->rtc_id.c_str(), inet_ntoa(sa.sin_addr), ntohs(sa.sin_port));
                connector->remote_addr = remote_addr;
            }
        } else {
            size = g_socket_receive_from(connector->sock, nullptr, (gchar*)buf, sizeof(buf), nullptr, &err);
            if (err != nullptr) {
                if (err->code != G_IO_ERROR_WOULD_BLOCK) {
                    dbgw("_rcm_connector_handle_recv(%s): %s", connector->rtc_id.c_str(), err->message);
                }
                g_error_free(err);
                return G_SOURCE_CONTINUE;
            }
        }

        if (is_rtcp(buf, size)) {
            _rcm_connector_handle_recv_rtcp(connector, buf, size);
        } else {
            _rcm_connector_handle_recv_rtp(connector, buf, size);
        }
    }
}

/**
 * 将 ssrc（placeholder） ---> sub rtc local ssrc
 * 发送
 *
 *
 * @param connector
 * @param buf
 * @param size
 * @return
 */
static gboolean _rcm_connector_handle_recv_rtcp(remote_connector *connector, const unsigned char *buf, int size) {
    connector->last_update_ms = get_timestamp_ms();

    auto ssrc = rtcp_get_ssrc(buf, size);

    int stream_type;

    if ((connector->peer_audio_ssrcs.size() > 0 && ssrc == connector->peer_audio_ssrcs[0])
        || (connector->peer_audio_ssrcs.size() > 1 && ssrc == connector->peer_audio_ssrcs[1])
        || ssrc == placeholder::audio.remote.pri
        || ssrc == placeholder::audio.remote.rtx) {
        stream_type = RTC_AUDIO;
    } else if ((connector->peer_video_ssrcs.size() > 0 && ssrc == connector->peer_video_ssrcs[0])
               || (connector->peer_video_ssrcs.size() > 1 && ssrc == connector->peer_video_ssrcs[1])
               || ssrc == placeholder::video.remote.pri
               || ssrc == placeholder::video.remote.rtx) {
        stream_type = RTC_VIDEO;
    } else {
        dbge("_rcm_connector_handle_recv_rtcp: remote connector '%s' could not handle ssrc '%u'", connector->rtc_id.c_str(), ssrc);
        return G_SOURCE_CONTINUE;
    }

    uint8_t pt = buf[1];
    if (pt == webrtc::RTCPUtility::PT_SR) {
        //dbgi("_rcm_connector_handle_recv_rtcp %s SR", stream_type == RTC_AUDIO ? "audio" : "video");
        for (auto it : connector->linkers) {
            auto linker = it.second;
            // 直接转发 publisher的RTCP
            rtc_endpoint_send_stream_data(linker->rtc, stream_type, rtc_codec_id_unknown, rtc_codec_id_unknown,
                                          1, 0, (unsigned char*)buf, size);
        }
    }

    return G_SOURCE_CONTINUE;
}

/**
 * 终端A--rtcp|rtp-> Xswitch1(pub_rtc) --socket:udp rtcp|rtp-> Xswitch2:_rcm_connector_handle_recv(sub_rtc) --rtcp|rtp->终端B
 * Xswitch1 将 rtcp pub rtc remote ssrc --> placeholder;  (process_recv_rtp_rtcp)
 *             rtp pub rtc remote ssrc --> pub rtc local ssrc (send_rtp_data)
 *
 *
 */
gboolean _rcm_connector_handle_recv_rtp(remote_connector *connector, const unsigned char *buf, int size) {

    connector->last_update_ms = get_timestamp_ms();

    auto ssrc = rtp_get_ssrc(buf, size);
    if (ssrc == 0) {
        dbge("_rcm_connector_handle_recv_rtp: remote connector '%s' could not handle received rtp, ssrc = %d", connector->rtc_id.c_str(), ssrc);
        return G_SOURCE_CONTINUE;
    }

    int stream_type;
    bool is_rtx = false;

    std::ptrdiff_t audio_ssrc_index = std::find(connector->peer_audio_ssrcs.begin(),
            connector->peer_audio_ssrcs.end(), ssrc) - connector->peer_audio_ssrcs.begin();
    std::ptrdiff_t video_ssrc_index = std::find(connector->peer_video_ssrcs.begin(),
            connector->peer_video_ssrcs.end(), ssrc) - connector->peer_video_ssrcs.begin();
    if (!connector->peer_audio_ssrcs.empty() && audio_ssrc_index < (std::ptrdiff_t)connector->peer_audio_ssrcs.size()) {
        stream_type = RTC_AUDIO;
        if (audio_ssrc_index % 2 != 0) {
            is_rtx = true;
        }
    } else if (!connector->peer_video_ssrcs.empty() && video_ssrc_index < (std::ptrdiff_t)connector->peer_video_ssrcs.size()) {
        stream_type = RTC_VIDEO;
        if (video_ssrc_index % 2 != 0) {
            is_rtx = true;
        }
    } else {
        //dbgw("remote connector '%s' received unknown ssrc '%u'",
        //        connector->rtc_id.c_str(), ssrc);
        return G_SOURCE_CONTINUE;
    }

    rtc_codec_id_t codec = rtc_codec_id_unknown;
    rtc_codec_id_t fec_codec = rtc_codec_id_unknown;

    auto pt = rtp_get_pt(buf, size);

    switch (pt) {
    case rtc_codec_default_pt_vp8:
        codec = rtc_codec_id_vp8;
        break;
    case rtc_codec_default_pt_h264:
        codec = rtc_codec_id_h264;
        break;
    case rtc_codec_default_pt_pcmu:
        codec = rtc_codec_id_pcmu;
        break;
    case rtc_codec_default_pt_pcma:
        codec = rtc_codec_id_pcma;
        break;
    case rtc_codec_default_pt_opus:
        codec = rtc_codec_id_opus;
        break;
    case rtc_codec_default_pt_ulpfec_unknown:
        codec = rtc_codec_id_ulpfec;
        fec_codec = rtc_codec_id_unknown;
        break;
    case rtc_codec_default_pt_ulpfec_vp8:
        codec = rtc_codec_id_ulpfec;
        fec_codec = rtc_codec_id_vp8;
        break;
    case rtc_codec_default_pt_ulpfec_h264:
        codec = rtc_codec_id_ulpfec;
        fec_codec = rtc_codec_id_h264;
        break;
    default:
        dbge("_rcm_connector_handle_recv_rtp: remote connector '%s' could not handle received rtp, pt = %d", connector->rtc_id.c_str(), pt);
        return G_SOURCE_REMOVE;
    }

    for (auto it : connector->linkers) {
        auto linker = it.second;
        rtc_endpoint_send_stream_data(linker->rtc, stream_type, codec, fec_codec,
                                      0, is_rtx ? 1 : 0, const_cast<unsigned char*>(buf), size);
    }

    return G_SOURCE_CONTINUE;
}

static void _rcm_connector_send_rtcp(remote_connector* connector, const unsigned char* buf, int size) {
    if (connector->remote_addr == nullptr) {
        dbgw("_rcm_connector_send_rtcp(%s) error: remote address not set", connector->rtc_id.c_str());
        return;
    }

    GError* err = nullptr;

    g_socket_send_to(connector->sock, connector->remote_addr, (const gchar*)buf, size, nullptr, &err);

    if (err != nullptr) {
        dbgw("_rcm_connector_send_rtcp(%s) error: %s", connector->rtc_id.c_str(), err->message);
        g_error_free(err);
    }
}

static remote_connector* rcm_get_connector(remote_connector_manager* rcm, const std::string& rtc_id) {

    auto it = rcm->connectors.find(rtc_id);

    if (it == rcm->connectors.end()) {

        auto rc = _rcm_new_connector(rtc_id.c_str());

        if (rc == nullptr) {
            return nullptr;
        }

        rcm->connectors.emplace(rtc_id, rc);

        return rc;
    }

    return it->second;
}

static void _rcm_linker_delete(remote_linker *linker) {
    dbgi("_rcm_linker_delete('%s')", rtc_endpoint_get_id(linker->rtc));

    // auto connector = linker->connector;

    if (linker->callback_link) {
        rtc_endpoint_unregister_callback(linker->rtc, linker->callback_link);
        linker->callback_link = nullptr;
    }

    delete(linker);
}

static void _rcm_linker_callback_rtc_destroy(rtc_endpoint_t rtc, void * context) {
    dbgi("_rcm_linker_callback_rtc_destroy");

    auto linker = static_cast<remote_linker*>(context);

    auto connector = linker->connector;

    auto linker_id = std::string(rtc_endpoint_get_id(rtc));

    connector->linkers.erase(linker_id);

    _rcm_linker_delete(linker);
    dbgi("_rcm_linker_callback_rtc_destroy end");
}

// todo: 这里只是转发subscriber发来的rtcp，由rtc_linker处理
/**
 * remote rtc，终端上行的rtcp
 *  -----》rtc_linker_recv_remote
 *
 *
 *
 * @param rtc
 * @param context
 * @param stream_type
 * @param codec
 * @param fec_codec
 * @param data_type
 * @param is_rtcp
 * @param is_rtx
 * @param buf
 * @param len
 * @return
 */
static bool _rcm_linker_callback_on_data(rtc_endpoint_t rtc, void * context, int stream_type, rtc_codec_id_t codec, rtc_codec_id_t fec_codec, int data_type, int is_rtcp, int is_rtx, unsigned char * buf, int len) {
    auto linker = static_cast<remote_linker*>(context);
    if (data_type != RTC_RECV_CLEAR) {
        return false;
    }

    if (!is_rtcp) {
        return false;
    }

    // 已经在 process_recv_rtp_rtcp 中将 sub rtc remote ssrc 改为了 placeholder
    _rcm_connector_send_rtcp(linker->connector, buf, len);
    return false;
}
